package com.idanchuang.count.service.impl;

import com.google.common.eventbus.EventBus;
import com.idanchuang.count.core.constant.UserCountEventTypeEnum;
import com.idanchuang.count.core.entity.KafkaSendEvent;
import com.idanchuang.count.service.AsyncCountCommandHandler;
import com.idanchuang.count.support.MetricCollector;
import com.idanchuang.user.count.service.api.request.command.BaseUpdateCommand;
import com.idanchuang.user.count.service.api.request.command.UserRelated;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

import static com.idanchuang.count.core.util.JacksonUtil.toJsonStr;

/**
 * @author Richard_yyf
 * @version 1.0 2021/5/31
 */
@Component
@Slf4j
public class AsyncCountCommandHandlerImpl implements AsyncCountCommandHandler {

    @Resource
    private EventBus kafkaSendEventBus;

    @Value("${count.topic-prefix}")
    private String topicPrefix;

    @Override
    public void publishCountEvent(UserCountEventTypeEnum userCountEventTypeEnum,
                                  BaseUpdateCommand command) {

        String topic = topicPrefix + "_" + userCountEventTypeEnum.getTag();
        String userCodeStr = null;
        if (command instanceof UserRelated) {
            userCodeStr = String.valueOf(((UserRelated) command).getUserCode());
        }

        // 如果userStr存在，会根据userCodeStr做分区
        ProducerRecord<Object, Object> record = userCodeStr == null ?
                new ProducerRecord<>(topic, toJsonStr(command)) : new ProducerRecord<>(topic, userCodeStr, toJsonStr(command));

        kafkaSendEventBus.post(new KafkaSendEvent(record, new ListenableFutureCallback<SendResult<Object, Object>>() {
            @Override
            public void onSuccess(SendResult<Object, Object> result) {
                MetricCollector.collectAfterEventSent(userCountEventTypeEnum.getTag());
            }

            @Override
            public void onFailure(Throwable ex) {
                log.error("发送消息至kafka失败, command: " + command.toString(), ex);
                MetricCollector.collectAfterEventSentFail(userCountEventTypeEnum.getTag());
            }
        }));
    }

}
